package cn.doitedu.api;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class _10_RichFunction_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        //
        List<String> list = Arrays.asList("a,a,a", "a,b,c", "a,c,c,d");
        DataStream<String> stream = env.fromCollection(list);

        // 变大写
        DataStream<String> uppered = stream.map(new RichMapFunction<String, String>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                // 一般是用来安排一些初始化逻辑
            }

            @Override
            public String map(String value) throws Exception {

                return value.toUpperCase();
            }

            @Override
            public void close() throws Exception {
                // task结束的时候会调用，可以安排一些资源关闭逻辑
            }
        });


        // 切单词打散
        DataStream<String> flatMapped = uppered.flatMap(new RichFlatMapFunction<String, String>() {

            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String word : split) {
                    out.collect(word);
                }
            }


        });


        // 变成  (w,1)
        DataStream<Tuple2<String, Integer>> tuples =
                flatMapped.map(s -> Tuple2.of(s, 1))
                        .returns(new TypeHint<Tuple2<String, Integer>>() {});

        // keyBy聚合
        tuples.keyBy(t -> t.f0)
                .sum("f1")
                .print("haha");


        env.execute();


    }


}
